package com.brillilab.message.consumer;

import com.aliyun.openservices.ons.api.Consumer;
import com.brillilab.common.rocketmq.OnsConfig;
import com.brillilab.common.rocketmq.OnsProperties;
import com.brillilab.message.handler.MessageHandler;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * 
 * @author lj
 *
 */
@Slf4j
@Configuration
public class OnsConsumerConfig {

	@Autowired
	private OnsConfig onsConfig;
	@Autowired
	private OnsProperties onsProperties;

	/**
	 * 启动webSocketAioTopic主题广播订阅消费者
	 * 当使用广播消费模式时，消息队列 RocketMQ 会将每条消息推送给集群内所有注册过的客户端，保证消息至少被每台机器消费一次。
	 * @return
	 */
	@Bean
	public Consumer aioTopicBroadcastConsumer() {
		Consumer consumer = onsConfig.getBroadcastConsumer(onsProperties.broadcastGroupId);
		consumer.subscribe(onsProperties.webSocketAioTopic, onsProperties.broadcastTag, new OnsBroadcastListener());
		consumer.start();
		log.info("{}  webSocketAioTopic主题广播订阅消费者创建完毕", "aioTopicBroadcastConsumer");
		return consumer;
	}

	
	/**
	 * 启动webSocketAioTopic主题集群订阅消费者 
	 * 当使用集群消费模式时，消息队列 RocketMQ 认为任意一条消息只需要被集群内的任意一个消费者处理即可。
	 * 
	 */
	@Bean
	public Consumer aioTopicClusteringConsumer() {
		Consumer consumer = onsConfig.getClusteringConsumer(onsProperties.clusteringGroupId);
		consumer.subscribe(onsProperties.appPushTopic, onsProperties.clusteringTag, new OnsClusteringListener());
		consumer.start();
		log.info("{}  appPushTopic主题集群订阅消费者创建完毕", "appTopicClusteringConsumer");
		return consumer;
	}

	@Bean
	public Consumer aioTopicClusteringCrmConsumer() {
		Consumer consumer = onsConfig.getClusteringConsumer(onsProperties.userCacheGroupId);
		consumer.subscribe(onsProperties.refreshUserCacheTopic, onsProperties.clusteringTag, new OnsClusteringCrmListener());
		consumer.start();
		log.info("{}  refreshUserCacheTopic主题集群订阅消费者创建完毕", "aioTopicClusteringCrmConsumer");
		return consumer;
	}

	@Autowired
	public void setMessageHandler(MessageHandler messageHandler){
		OnsBroadcastListener.messageHandler = messageHandler;
		OnsClusteringListener.messageHandler = messageHandler;
		OnsClusteringCrmListener.messageHandler=messageHandler;
	}

}
